package com.sheep.hadoop.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.net.URI;

/**
 * 该方法用于指定一个job任务
 * MapReduce 启动任务
 *
 * @author wangze
 * @date 2021-02-28 16:05.
 **/
public class JobMain extends Configured implements Tool {

	private final static String HDFS_PATH = "hdfs://192.168.168.101:9000";

	@Override
	public int run(String[] strings) throws Exception {
		// 1. 创建一个job任务对象
		// super.getConf()
		// 获取的是main中新建的配置类，这里配置类必须一致
		// 这边获取config，只能从父类获取
		Job job = Job.getInstance(super.getConf(), "wordCount");
		// 设置job处理类
		job.setJarByClass(JobMain.class);

		// 2. 配置一个任务对象
		// 第一步：指定文件的读取方式和读取路径
		job.setInputFormatClass(TextInputFormat.class);
		// 设置文件路径，因为会读取所有文件
		FileInputFormat.addInputPath(job, new Path(HDFS_PATH+"/wordcount"));
		// 第二步；指定Map阶段的处理方式和数据类型
		job.setMapperClass(WordCountMapper.class);
		// 设置map k2  类型
		job.setMapOutputKeyClass(Text.class);
		// 设置map v3类型
		job.setMapOutputValueClass(LongWritable.class);
		// 第三四五六 默认 因为是shuttle，暂时不做处理
		// 第七步 ： Reduce阶段的处理方式和处理类型
		job.setReducerClass(WordCountReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.class);
		// 第八步： 设置输出类型
		job.setOutputFormatClass(TextOutputFormat.class);

		// 设置combiner
		job.setCombinerClass(WordCountReducer.class);

		// 设置输出路径
		TextOutputFormat.setOutputPath(job, new Path(HDFS_PATH+"/wordcount1"));

		//等待任务结束
		boolean isComplete = job.waitForCompletion(true);
		return isComplete ? 0 : 1;
	}

	public static void main(String[] args) throws Exception {
		// configuration 是公用的 可以看ToolRunner.run的代码会设置JobMain的configuration
		Configuration configuration = new Configuration();

		// 删除创建的文件
		// 注意 hadoop 有权限的问题，因为我虚拟机使用root创建hadoop的，所以这么传root

		//删除已存在的输出目录
		FileSystem fileSystem = FileSystem.get(configuration);
		if(fileSystem.exists(new Path(HDFS_PATH+"/wordcount1"))){
			//递归删除wordcount1文件夹
			fileSystem.delete(new Path(HDFS_PATH+"/wordcount1"),true);
			System.out.println("文件删除成功");
		}



		// 返回0 任务成功
		int run = ToolRunner.run(configuration, new JobMain(), args);
		System.exit(run);
	}
}
